package com.rxjava;

import java.util.concurrent.TimeUnit;

import io.reactivex.Observable;
import io.reactivex.functions.Function;
import io.reactivex.schedulers.Schedulers;

public class RetryWithDelay implements Function<Observable<? extends Throwable>, Observable<?>> {

	private final int maxRetries;
	private final int retryDelayMillis;
	private int retryCount;

	public RetryWithDelay(int maxRetries, int retryDelayMillis) {
		this.maxRetries = maxRetries;
		this.retryDelayMillis = retryDelayMillis;
	}

	@Override
	public Observable<?> apply(Observable<? extends Throwable> attempts) {
		return attempts.flatMap(new Function<Throwable, Observable<?>>() {
			@Override
			public Observable<?> apply(Throwable throwable) {
				if (++retryCount <= maxRetries) {
					// When this Observable calls onNext, the original Observable will be retried (i.e. re-subscribed).
					System.out.println("get error, it will try after " + retryDelayMillis + " millisecond, retry count " + retryCount);
					return Observable.timer(retryDelayMillis, TimeUnit.MILLISECONDS, Schedulers.trampoline());
				}
				// Max retries hit. Just pass the error along.
				return Observable.error(throwable);
			}
		});
	}
}